package producer

import (
	"errors"
	"fmt"
	"github.com/alecthomas/log4go"
	"github.com/streadway/amqp"
	"time"
)

type MQProducer interface {
	Start() error
	Publish(body []byte) (err error)
	Stop() error
}

func New(config Config) MQProducer {
	return &producer{
		Config:      config,
		connOk:      false,
		notifyClose: make(chan *amqp.Error),
		closeChan:   make(chan bool),
	}
}

type Config struct {
	Uri          string //连接参数  amqp://guest:guest@localhost:5672/
	ExchangeName string //交换机名称
	ExchangeType string //交换机类型 direct|fanout|topic|x-custom
	RoutingKey   string //路由key
}

type producer struct {
	Config
	/***********/
	conn        *amqp.Connection //连接
	channel     *amqp.Channel    //通道
	notifyClose chan *amqp.Error
	closeChan   chan bool
	connOk      bool //连接状态
}

//
//func (this *producer) Name() string {
//	return "rabbitmq producer"
//}
//func (this *producer) Init() {
//	var err error
//	if this.uri, err = config.GetStringDefault("rabbitmq", "uri", "amqp://guest:guest@localhost:5672/"); err != nil {
//		panic(fmt.Sprintf("%s，获取rabbitmq配置'uri'失败:%s", this.Name(),err.Error()))
//	}
//	if this.exchangeName,err=config.GetStringDefault("rabbitmq", "exchangeName", "test-exchange");err!=nil{
//		panic(fmt.Sprintf("%s，获取rabbitmq配置'exchangeName'失败:%s",this.Name(), err.Error()))
//	}
//	if this.exchangeType,err=config.GetStringDefault("rabbitmq","exchangeType","direct");err!=nil{
//		panic(fmt.Sprintf("%，获取rabbitmq配置'exchangeType'失败:%s",this.Name(), err.Error()))
//	}
//	if this.routingKey,err=config.GetStringDefault("rabbitmq","routingKey","test-key");err!=nil{
//		panic(fmt.Sprintf("%s，获取rabbitmq配置'routingKey'失败:%s",this.Name(), err.Error()))
//	}
//	this.connOk = false
//	this.notifyClose = make(chan *amqp.Error)
//	this.closeChan = make(chan bool)
//}

//开启，会阻塞
func (this *producer) Connect() error {
	//监听连接异常
	go func() {
		this.begin()
	}()
	return nil
}

//关闭
func (this *producer) Start() error {
	if !this.connOk {
		return errors.New(fmt.Sprintf("rabbitmq producer,已关闭!"))
	}
	//关闭连接的同时，channel也会关闭
	err := this.conn.Close()
	if err != nil {
		return err
	}
	this.connOk = false
	close(this.closeChan)
	return nil
}

//关闭
func (this *producer) Stop() error {
	if !this.connOk {
		return errors.New(fmt.Sprintf("rabbitmq producer,已关闭"))
	}
	//关闭连接的同时，channel也会关闭
	err := this.conn.Close()
	if err != nil {
		return err
	}
	this.connOk = false
	close(this.closeChan)
	return nil
}

//推送消息
func (this *producer) Publish(body []byte) (err error) {
	if !this.connOk {
		return errors.New(fmt.Sprintf("rabbitmq producer,连接状态不正常，暂不能发送消息"))
	}
	if err = this.channel.Publish(
		this.ExchangeName, // publish to an exchange
		this.RoutingKey,   // routing to 0 or more queues
		false,             // mandatory
		false,             // immediate
		amqp.Publishing{
			Headers:         amqp.Table{},
			ContentType:     "text/plain",
			ContentEncoding: "",
			Body:            body,
			DeliveryMode:    amqp.Transient, // 1=non-persistent, 2=persistent
			Priority:        0,              // 0-9
			// a bunch of application/implementation-specific fields
		},
	); err != nil {
		log4go.Error("rabbitmq producer,生产者推送消息异常: %s", err.Error())
		//this.errChan<-err
		return err
	}
	log4go.Info("rabbitmq producer,生产者消息发送成功")
	return nil
}

//重试连接
func (this *producer) begin() {
	log4go.Info("rabbitmq producer开启异常监听")
	for {
		this.connOk = false
		for {
			if err := this.connect(); err != nil {
				log4go.Error("rabbitmq producer,创建新连接失败：%s,准备重试", err.Error())
				time.Sleep(10 * time.Second)
				continue
			} else {
				this.notifyClose = make(chan *amqp.Error)
				this.channel.NotifyClose(this.notifyClose)
				break
			}
		}
		select {
		case <-this.closeChan:
			return
		case <-this.notifyClose:
			break
		}
	}
}

//创建连接
func (this *producer) connect() (err error) {
	if this.connOk {
		return errors.New("已有连接并且连接正常，无需重复创建")
	}
	this.conn, err = amqp.Dial(this.Uri)
	if err != nil {
		log4go.Error("rabbitmq producer,创建rabbitmq连接失败: %s", err.Error())
		return err
	} else {
		log4go.Info("rabbitmq producer,创建rabbitmq连接成功")
	}
	//创建channel
	this.channel, err = this.conn.Channel()
	if err != nil {
		log4go.Error("rabbitmq producer,获取channel失败: %s", err.Error())
		return err
	}
	//创建交换机
	if err = this.exchangeDeclare(); err != nil {
		log4go.Error("rabbitmq producer,创建exchange失败: %s", err.Error())
		return err
	}

	this.connOk = true
	return nil
}

//创建交换机
func (this *producer) exchangeDeclare() (err error) {
	if err = this.channel.ExchangeDeclare(
		this.ExchangeName, // name
		this.ExchangeType, // type
		true,              // durable
		false,             // auto-deleted
		false,             // internal
		false,             // noWait
		nil,               // arguments
	); err != nil {
		log4go.Error("rabbitmq producer,创建exchange失败: %s", err.Error())
		return err
	}
	return nil
}
